package com.lx.kafka.consumer;

import com.lx.kafka.Properties.ConsumerProperties;
import com.lx.kafka.log.KafkaLog;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Set;

/**
 * Description:
 * Copyright:   Copyright (c)2019
 * Company:     zefu
 *
 * @author: 张李鑫
 * @version: 1.0
 * Create at:   2022-02-20 22:32:58
 * <p>
 * Modification History:
 * Date         Author      Version     Description
 * ------------------------------------------------------------------
 * 2022-02-20     张李鑫                     1.0         1.0 Version
 */
public class Consumer {

    public static void main(String[] args) {
        KafkaLog.closeLog();
        ConsumerProperties consumerProperties = new ConsumerProperties();
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProperties.properties);
        ArrayList<String> strings = new ArrayList<>();
        strings.add("topic02");
        consumer.subscribe(strings);
        //指定位置开始消费
        Set<TopicPartition> assignment = consumer.assignment();
        for (TopicPartition topicPartition : assignment) {
            consumer.seek(topicPartition, 10);
        }
        while (true) {
            ConsumerRecords<String, String> message = consumer.poll(Duration.ofSeconds(1));
            if (!message.isEmpty()) {
                message.forEach(System.out::println);
            }
        }

    }
}
